 1.Kafka高级特性解析之消费者中消息接收中消费者位移管理
   
   Kafka中，消费者根据消息的位移顺序消费消息。
   消费者的位移由消费者管理，可以存储于zookeeper中，也可以存储于Kafka主题__consumer_offsets中。
   Kafka提供了消费者API，让消费者可以管理自己的位移。
   API如下：KafkaConsumer<K, V>
   项目   细节
   API    public void assign(Collection<TopicPartition> partitions)
   说明   给当前消费者手动分配一系列主题分区。
         手动分配分区不支持增量分配，如果先前有分配分区，则该操作会覆盖之前的分配。
         如果给出的主题分区是空的，则等价于调⽤unsubscribe方法。
		 手动分配主题分区的方法不使用消费组管理功能。当消费组成员变了，
		 或者集群或主题的元数据改变了，不会触发分区分配的再平衡。
		 手动分区分配assign(Collection)不能和⾃动分区分配subscribe(Collection,
		 ConsumerRebalanceListener)一起使用。
		 如果启用了自动提交偏移量，则在新的分区分配替换旧的分区分配之前，
		 会对旧的分区分配中的消费偏移量进行异步提交。
   API   public Set<TopicPartition> assignment()
   说明  获取给当前消费者分配的分区集合。如果订阅是通过调用assign方法直接分配主题分区，则返回相同的集
         合。如果使用了主题订阅，该方法返回当前分配给该消费者的主题分区集合。如果分区订阅还没开始进行4
		 分区分配，或者正在重新分配分区，则会返回none。
   API   public Map<String, List<PartitionInfo>> listTopics()
   说明  获取对⽤户授权的所有主题分区元数据。该方法会对服务器发起远程调用。
   API   public List<PartitionInfo> partitionsFor(String topic)
   说明  获取指定主题的分区元数据。如果当前消费者没有关于该主题的元数据，就会对服务器发起远程调用。
   API   public Map<TopicPartition, Long> beginningOffsets(Collection<TopicPartition> partitions)
   说明  对于给定的主题分区，列出它们第一个消息的偏移量。
         注意，如果指定的分区不存在，该方法可能会永远阻塞。
         该方法不改变分区的当前消费者偏移量。
   API   public void seekToEnd(Collection<TopicPartition> partitions)
   说明  将偏移量移动到每个给定分区的最后⼀个。
         该方法延迟执行，只有当调用过poll方法或position方法之后才可以使用。
         如果没有指定分区，则将当前消费者分配的所有分区的消费者偏移量移动到最后。
		 如果设置了隔离级别为：isolation.level=read_committed，则会将分区的消费偏移量移
		 动到最后一个稳定的偏移量，即下一个要消费的消息现在还是未提交状态的事务消息。
   API   public void seek(TopicPartition partition, long offset)
   说明  将给定主题分区的消费偏移量移动到指定的偏移量，即当前消费者下一条要消费的消息偏移量。
         若该方法多次调用，则最后一次的覆盖前⾯的。
         如果在消费中间随意使用，可能会丢失数据。
   API   public long position(TopicPartition partition)
   说明  检查指定主题分区的消费偏移量
   API   public void seekToBeginning(Collection<TopicPartition> partitions)
   说明  将给定每个分区的消费者偏移量移动到它们的起始偏移量。该方法懒执行，
         只有当调用过poll方法或position方法之后才会执行。如果没有提供分区
		 则将所有分配给当前消费者的分区消费偏移量移动到起始偏移量
   
   1).准备数据
   # 生成消息文件件
[root@linux121 ~]# for i in `seq 60`; do echo "hello lagou $i" >> messages.txt; done
   # 创建主题，三个分区，每个分区一个副本
kafka-topics.sh --zookeeper linux121:2181/myKafka --create --topic tp_demo_01 --partitions 3 
--replication-factor 1   
   # 将消息生产到主题中
[root@linux121 ~]# kafka-console-producer.sh --broker-list linux121:9092 --topic tp_demo_01 <
messages.txt
[root@linux121 ~]# kafka-console-consumer.sh --bootstrap-server linux121:9092 --topic
tp_demo_01 --from-beginning
   2).API实战
   com.lagou.kafka.demo.MyOffsetManager
package com.lagou.kafka.demo;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.*;
import java.util.function.BiConsumer;

public class MyOffsetManager {
    public static void main(String[] args) {

        Map<String, Object> configs = new HashMap<String, Object>();
        configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "linux121:9092");
        configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        // group.id很重要
        configs.put(ConsumerConfig.GROUP_ID_CONFIG, "mygrp1");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(configs);
//        consumer.subscribe(Collections.singleton("tp_demo_01"));
        // 如何手动给消费者分配分区?
        // 1.需要知道有哪些主题可以访问，和消费
        // 获取当前消费者可以访问和消费的主题以及他们的分区信息
//        final Map<String, List<PartitionInfo>> stringListMap = consumer.listTopics();
//        stringListMap.forEach(new BiConsumer<String, List<PartitionInfo>>() {
//            @Override
//            public void accept(String topicName, List<PartitionInfo> partitionInfos) {
//                System.out.println("主题名称：" + topicName);
//                for (PartitionInfo partitionInfo : partitionInfos) {
//                    System.out.println(partitionInfo);
//
//                }
//
//            }
//        });
        final Set<TopicPartition> assignment1 = consumer.assignment();
        for (TopicPartition partition : assignment1) {
            System.out.println(partition);
        }
        System.out.println("------------------");
        // 给当前消费者分配指定的主题分区
        consumer.assign(Arrays.asList(
                new TopicPartition("tp_demo_01", 0),
                new TopicPartition("tp_demo_01", 1),
                new TopicPartition("tp_demo_01", 2)
        ));
        
        // 获取给当前消费者分配指定的主题分区
//        final Set<TopicPartition> assignment = consumer.assignment();
//        for (TopicPartition partition : assignment) {
//            System.out.println(partition);
//        }
        // 查看当前消费者在指定主题分区上的消费者偏移量
//        final long offset0 = consumer.position(new TopicPartition("tp_demo_01", 0));
//        System.out.println("当前主题在0号分区上的位移：" + offset0);

//        consumer.seekToBeginning(Arrays.asList(
//                new TopicPartition("tp_demo_01", 0),
//                new TopicPartition("tp_demo_01", 2)
//        ));

        long offset0 = consumer.position(new TopicPartition("tp_demo_01", 0));
        long offset1 = consumer.position(new TopicPartition("tp_demo_01", 1));
        long offset2 = consumer.position(new TopicPartition("tp_demo_01", 2));
        System.out.println(offset0);
        System.out.println(offset1);
        System.out.println(offset2);

//        consumer.seekToEnd(Arrays.asList(new TopicPartition("tp_demo_01", 2)));
//        offset0 = consumer.position(new TopicPartition("tp_demo_01", 0));
//        offset1 = consumer.position(new TopicPartition("tp_demo_01", 1));
//        offset2 = consumer.position(new TopicPartition("tp_demo_01", 2));
//        System.out.println(offset0);
//        System.out.println(offset1);
//        System.out.println(offset2);

        consumer.seek(new TopicPartition("tp_demo_01", 2), 14);
        offset0 = consumer.position(new TopicPartition("tp_demo_01", 0));
        offset1 = consumer.position(new TopicPartition("tp_demo_01", 1));
        offset2 = consumer.position(new TopicPartition("tp_demo_01", 2));
        System.out.println(offset0);
        System.out.println(offset1);
        System.out.println(offset2);
        consumer.close();
    }
} 
    com.lagou.kafka.demo.MyConsumer
package com.lagou.kafka.demo;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;

public class MyConsumer {
    public static void main(String[] args) {
        Map<String, Object> configs = new HashMap<String, Object>();
        configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "linux121:9092");
        configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        // group.id很重要
        configs.put(ConsumerConfig.GROUP_ID_CONFIG, "mygrp1");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(configs);

        consumer.subscribe(Arrays.asList("tp_demo_01"));
        while (true) {
            final ConsumerRecords<String, String> records = consumer.poll(1_000);
            records.forEach(new Consumer<ConsumerRecord<String, String>>() {
                @Override
                public void accept(ConsumerRecord<String, String> record) {
                    System.out.println(record);
                }
            });
        }
    }
}

    